package com.jy.oms.framework.asyn.task.executor;

import cn.hutool.core.thread.ThreadUtil;
import com.jy.oms.framework.asyn.task.collector.TimeoutTaskCollector;
import com.jy.oms.framework.asyn.task.constants.TaskQueueTypeEnum;
import com.jy.oms.framework.asyn.task.exception.TaskConfigException;
import com.jy.oms.framework.asyn.task.executor.bo.TaskPollResult;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.model.TaskGroup;
import com.jy.oms.framework.asyn.task.model.TaskHandlerWrapper;
import com.jy.oms.framework.asyn.task.model.TaskWrapper;
import com.jy.oms.framework.asyn.task.queue.TaskQueueManagerProxy;
import com.jy.oms.framework.thread.CloseableThread;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 任务组控制器
 *
 * @author liuquan
 */
@Service
public class GroupTaskController implements ITaskController {
    protected static Logger logger = LoggerFactory.getLogger(GroupTaskController.class);
    @Autowired
    protected ITaskHandlerManager taskHandlerManager;
    @Autowired
    protected TimeoutTaskCollector timeoutTaskCollector;
    @Autowired
    private ITaskExecutor taskExecutor;
    @Autowired
    private List<ITaskControllerFilter> taskControlFilterList;
    @Autowired
    private List<ITaskExtractor> taskExtractorList;

    @Override
    public void init(TaskGroup taskGroup) {
        //校验分组核心线程数是否一致
        Map<TaskGroup, List<TaskHandlerWrapper>> taskHandlerMap = taskHandlerManager.loadAllExecutableTaskHandler();
        List<Integer> taskGroupCoreThreadNums = taskHandlerMap.get(taskGroup).stream().map(item -> {
            Integer taskGroupCoreThreadNum = item.getExecuteStrategy().getTaskGroupCoreThreadNum();
            return taskGroupCoreThreadNum == null || taskGroupCoreThreadNum < 0 ? 0 : taskGroupCoreThreadNum;
        }).distinct().collect(Collectors.toList());
        if (taskGroupCoreThreadNums.size() > 1) {
            throw new TaskConfigException("任务组【" + taskGroup.getGroup() + "】所有任务设置的核心线程数不相同");
        }

        if (taskGroup.getTaskQueueType() == TaskQueueTypeEnum.DB) {
            //校验分组队列表名是否一致
            List<String> taskGroupDbQueueNames = taskHandlerMap.get(taskGroup).stream().map(item -> {
                String taskGroupDbQueueName = item.getExecuteStrategy().getTaskGroupDbQueueName();
                return taskGroupDbQueueName == null ? "" : taskGroupDbQueueName;
            }).distinct().collect(Collectors.toList());
            if (taskGroupDbQueueNames.size() > 1) {
                throw new TaskConfigException("任务组【" + taskGroup.getGroup() + "】所有任务设置的队列表名不相同");
            }
        }
    }

    @Override
    public void schedule(TaskGroup taskGroup) {
        //任务执行器初始化
        taskExecutor.init(taskGroup);

        //任务调度线程
        Thread groupPerConsumerThread = new GroupTaskControllerThread(taskGroup);
        groupPerConsumerThread.setName("GroupTaskControllerThread#" + taskGroup.getGroup());
        groupPerConsumerThread.start();
    }

    private ITaskExtractor getTaskExtractor(TaskGroup taskGroup) {
        for (ITaskExtractor taskExtractor : taskExtractorList) {
            if (taskExtractor.ableExtract(taskGroup)) {
                return taskExtractor;
            }
        }

        throw new TaskConfigException("未找到任务抽取器");
    }

    /**
     * 当任务执行失败时，mem/redis 任务数据需重新加入队列
     */
    private void handleCacheTaskOnExecuteFailure(TaskPollResult taskPollResult) {
        if (taskPollResult == null || (CollectionUtils.isEmpty(taskPollResult.getExclusiveTasks())
                && CollectionUtils.isEmpty(taskPollResult.getTimeoutTasks()))) {
            return;
        }

        if (CollectionUtils.isNotEmpty(taskPollResult.getExclusiveTasks())) {
            for (TaskWrapper taskWrapper : taskPollResult.getExclusiveTasks()) {
                taskWrapper.getTaskDto().setDelayedSeconds(60L);
                TaskQueueManagerProxy.locate(taskWrapper.getTaskCode()).add(taskWrapper);
            }
        }

        if (CollectionUtils.isNotEmpty(taskPollResult.getTimeoutTasks())) {
            for (TaskWrapper taskWrapper : taskPollResult.getTimeoutTasks()) {
                taskWrapper.getTaskDto().setDelayedSeconds(60L * 60);
                TaskQueueManagerProxy.locate(taskWrapper.getTaskCode()).add(taskWrapper);
            }
        }
    }

    private class GroupTaskControllerThread extends CloseableThread {
        private TaskGroup taskGroup;
        private int taskEmptyQueryNum;

        public GroupTaskControllerThread(TaskGroup taskGroup) {
            this.taskGroup = taskGroup;
        }

        @Override
        public void run() {
            ThreadUtil.sleep(15, TimeUnit.SECONDS);
            while (!stoped) {
                logger.info("异步任务执行开始, taskGroup: {}", taskGroup);
                try {
                    this.schedule();
                } catch (Exception e) {
                    if (!stoped) {
                        logger.error("调度任务报错", e);
                    }
                }
            }
        }

        @Override
        protected void afterStop() {
            Thread.currentThread().interrupt();
        }

        private void schedule() {
            boolean checkResult = taskControlFilterList.stream()
                    .allMatch(f -> f.checkTaskGroupAbleRun(taskGroup));
            if (!checkResult) {
                logger.info("异步任务跳过，当前服务器无法执行任务, taskGroup: {}", taskGroup);
                ThreadUtil.sleep(60, TimeUnit.SECONDS);
                return;
            }

            boolean ableExecute = taskExecutor.isAbleExecute(taskGroup);
            if (!ableExecute) {
                logger.warn("异步任务跳过，当前组线程池繁忙无法执行任务, taskGroup: {}", taskGroup);
                ThreadUtil.sleep(60, TimeUnit.SECONDS);
                return;
            }

            ITaskExtractor taskExtractor = getTaskExtractor(taskGroup);
            TaskPollResult taskPollResult = taskExtractor.extract(taskGroup);
            if (taskPollResult != null && CollectionUtils.isNotEmpty(taskPollResult.getExecutableTasks())) {
                taskEmptyQueryNum = 0;
                taskExecutor.executeTask(taskGroup, taskPollResult.getExecutableTasks());
            } else {
                if ((taskGroup.getTaskQueueType() == TaskQueueTypeEnum.REDIS
                        || taskGroup.getTaskQueueType() == TaskQueueTypeEnum.MEM)) {
                    handleCacheTaskOnExecuteFailure(taskPollResult);
                }

                taskEmptyQueryNum++;
                if (taskEmptyQueryNum >= 3) {
                    ThreadUtil.sleep(60, TimeUnit.SECONDS);
                } else {
                    ThreadUtil.sleep(10, TimeUnit.SECONDS);
                }
            }
        }
    }

}
